package com.shujia.ads

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.catalog.hive.HiveCatalog

/*
对每条微博进行评分
评分： comments_count + reposts_count +  attitudes_count
 */
object Demo3ComputeWeiBoTopNIndex {
  def main(args: Array[String]): Unit = {
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()

    val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)

    val hiveCatalog = new HiveCatalog("myHive", "sent", "SentimentCompute/src/main/resources")

    bsTableEnv.registerCatalog("myHive", hiveCatalog)
    bsTableEnv.useCatalog("myHive")

    bsTableEnv.executeSql(
      """
        |insert into ads.ads_mysql_weibo_topN
        |select
        |id,
        |CONCAT('https://m.weibo.cn/detail/',id) as url,
        |text,
        |(comments_count+reposts_count+attitudes_count) as score
        |from dwd.dwd_kafka_weibo_msk
        |
      """.stripMargin)

    bsEnv.execute()

  }
}
